package com.els.connection;

import java.io.IOException;
import java.io.InputStream;
import java.net.InetAddress;
import java.util.List;
import java.util.Properties;


import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.settings.Settings.Builder;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.shield.ShieldPlugin;

import com.els.ElsConnection;
import com.google.gson.JsonObject;

public class ConfigElastic extends AbstractElastic implements ElsConnection{
	private String configPath;
	/**
	 * configPath中的参数包括，
	 * es.url			els的地址，用分号隔开
	 * cluster.name		集群名称
	 * shield.user	（可选）shield用户及密码，格式为username:password
	 * @param configPath
	 */
	public ConfigElastic(String configPath){
		this.configPath=configPath;
		init();	//初始化参数
	}
	private void init(){
		Properties prop =  new  Properties();    
        InputStream in = getClass().getResourceAsStream( configPath ); 
        try  {    
            prop.load(in);    
            esUrl = prop.getProperty( "es.url" ).trim(); 
            cluster = prop.getProperty( "cluster.name" ).trim(); 
            shieldUser=prop.getProperty( "shield.user" )==null?null:prop.getProperty( "shield.user" ).trim();
            //logger
            logger.debug("init success!");
            logger.debug("configPath:"+configPath);
            logger.debug("esUrl:"+esUrl);
            logger.debug("cluster:"+cluster);
            logger.debug("shieldUser:"+shieldUser);
        }catch(Exception e){
        	logger.error("Load config.properties has an error!");
        	e.printStackTrace();
        }finally{
        	try {
				in.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
        }
	}
	/**
	 * 创建client，建立连接
	 */
	public boolean buildClient() {
		if(client!=null)return true;
		Builder builder = Settings.settingsBuilder()
		        .put("cluster.name", cluster);
		if(shieldUser!=null){
			builder.put("shield.user", shieldUser);
		}
		Settings settings = builder.build();
		String[] urls=esUrl.split(";");
		try{
			TransportClient tClient=TransportClient.builder()
					.addPlugin(ShieldPlugin.class)
					.settings(settings).build();
			for (String url : urls) {
				String ip=url.split(":")[0];
				String port=url.split(":")[1];
				tClient.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ip),Integer.parseInt(port)));
		
			}
			client=tClient;
			logger.info("BuildClient "+esUrl+" success!");
			return true;
		}catch(Exception e){

			return false;
		}
	}
	/**
	 * @param index 索引名
	 * @param type 类型名
	 * @param jsonList 导入数据列表，json格式
	 * @param id 导入的id命名取值，当为null时，则由ELS自动生成
	 * @return int为批量成功的数量
	 */
	public int singleBulkIndex(String index, String type,
			List<JsonObject> jsonList, String id) {
		int fromIndex=0;
		int toIndex=_buffer;
		int numberInsert=0;
		while (fromIndex<jsonList.size()) {
			if(toIndex>jsonList.size()){
				toIndex=jsonList.size();
			}
			List<JsonObject> bufferList=jsonList.subList(fromIndex, toIndex);
			if(bulkIndex(index, type, bufferList, id)){
				numberInsert+=(toIndex-fromIndex);
			}
			fromIndex+=_buffer;
			toIndex+=_buffer;
		}
		return numberInsert;
	}
	

	public int singleBulkIndex(String index, String type,
			List<JsonObject> jsonList, String id, String parent) {
		// TODO Auto-generated method stub
		int fromIndex=0;
		int toIndex=_buffer;
		int numberInsert=0;
		while (fromIndex<jsonList.size()) {
			if(toIndex>jsonList.size()){
				toIndex=jsonList.size();
			}
			List<JsonObject> bufferList=jsonList.subList(fromIndex, toIndex);
			if(bulkIndex(index, type, bufferList, id, parent)){
				numberInsert+=(toIndex-fromIndex);
			}
			fromIndex+=_buffer;
			toIndex+=_buffer;
		}
		return numberInsert;
	}
	/**
	 * @param <T>泛型，可为IndexRequestBuilder,DeleteRequestBuilder,UpdateRequestBuilder
	 * @param requestList bulk的request列表
	 * @return int为批量成功的数量
	 */
	public <T> int singleBulk(List<T> requestList) {
		int fromIndex=0;
		int toIndex=_buffer;
		int numberInsert=0;
		while (fromIndex<requestList.size()) {
			if(toIndex>requestList.size()){
				toIndex=requestList.size();
			}
			List<T> bufferList=requestList.subList(fromIndex, toIndex);
			if(bulk(bufferList)){
				numberInsert+=(toIndex-fromIndex);
			}
			fromIndex+=_buffer;
			toIndex+=_buffer;
		}
		return numberInsert;
	}
	

}
